package com.storm.demo;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

import java.util.Map;
import java.util.Random;

/**
 * =================================
 * <p>
 * Created by cjj on 2018/10/23.
 * <p>
 * 描述: 获取输入流spout
 */

public class RandomWordSpout extends BaseRichSpout {

    private SpoutOutputCollector  spoutOutputCollector;
    //模拟数据，一般从消息队列中获取数据

    String[] words = {"scala","spark","jersy","netty","tomcat","nutch","storm"};

    /**
     *  初始化方法,在spout组件初始化的时候调用一次
     * @param map
     * @param topologyContext
     * @param spoutOutputCollector
     */
    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.spoutOutputCollector = spoutOutputCollector;
    }

    /**
     * 主要逻辑代码
     * 不断的往下一个组件发送tuple消息
     * 这是spout的核心代码
     */
    @Override
    public void nextTuple() {

        Random random = new Random();
        int i = random.nextInt(words.length);
        //随机拿到一个单次
        String word = words[i];

        //将拿到的东西封装为tuple，发送消息给下一个组件
        spoutOutputCollector.emit(new Values(word));

        //
        Utils.sleep(500);

    }

    /**
     * 声明输出字段
     * @param outputFieldsDeclarer
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("wordName"));
        //还可以定义流 不用会设置默认流
        //outputFieldsDeclarer.declareStream();
    }
}
